package cn.com.sparkle.paxos;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.NetworkInterface;
import java.net.SocketException;
import java.util.Enumeration;
import java.util.HashSet;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.TimeUnit;

import cn.com.sparkle.paxos.addprocess.AddRequestDealer;
import cn.com.sparkle.paxos.addprocess.speedcontrol.SpeedControlModel;
import cn.com.sparkle.paxos.deamon.CatchUpDeamon;
import cn.com.sparkle.paxos.deamon.CheckMasterSenatorStateDeamon;
import cn.com.sparkle.paxos.deamon.HeatBeatCheckDeamon;
import cn.com.sparkle.paxos.deamon.InstanceExecutor;
import cn.com.sparkle.paxos.deamon.ReConnectThread;
import cn.com.sparkle.paxos.event.EventsManager;
import cn.com.sparkle.paxos.net.ClientNetHandler;
import cn.com.sparkle.paxos.net.ServerNetHandler;
import cn.com.sparkle.paxos.userserver.UserHandlerInterface;
import cn.com.sparkle.paxos.userserver.UserServerHandler;
import cn.com.sparkle.raptor.core.handler.IoHandler;
import cn.com.sparkle.raptor.core.protocol.MultiThreadProtecolHandler;
import cn.com.sparkle.raptor.core.protocol.javaobject.ObjectProtocol;
import cn.com.sparkle.raptor.core.transport.socket.nio.NioSocketClient;
import cn.com.sparkle.raptor.core.transport.socket.nio.NioSocketConfigure;
import cn.com.sparkle.raptor.core.transport.socket.nio.NioSocketServer;

public class ClassicPaxosServer {
	private Configuration conf;
	private NioSocketServer server;
	private EventsManager eventsManager = new EventsManager();
	private ClusterState cState;
	private ReConnectThread reConnectThread = new ReConnectThread();

	public void init(String filePath, UserHandlerInterface userHandlerInterface)
			throws NumberFormatException, Exception {
		this.conf = new Configuration(filePath, eventsManager);
		cState = new ClusterState(eventsManager, conf);
		ServerNetHandler handler = tryBind();
		InstanceExecutor ie = new InstanceExecutor(userHandlerInterface, conf,
				eventsManager);
		// initiate account book
		AccountBook aBook = new AccountBook(conf);
		// initiate instance executor thread
		ie.start(aBook);
		aBook.initInstanceExecutor(ie);
		// initialize server handler
		handler.setaBook(aBook);
		handler.setcState(cState);
		// handler.setAddRequestDealer(new AddRequestDealer(cState, conf, aBook,
		// eventsManager));
		// initialize speed control
		SpeedControlModel speedControlModel = new SpeedControlModel(
				eventsManager, conf);
		// connect all node
		ClientNetHandler phandler = connectNodes();
		// initiate heart beat deamon
		HeatBeatCheckDeamon beatCheckDeamon = new HeatBeatCheckDeamon(cState,
				eventsManager);
		beatCheckDeamon.start();
		// initiate master check deamon
		Thread checkSenatorDeamon = new Thread(
				new CheckMasterSenatorStateDeamon(cState, conf, eventsManager));
		checkSenatorDeamon.setName("checkSenatorDeamon");
		checkSenatorDeamon.start();
		// initiate catch up deamon
		Thread catchUpDeamon = new Thread(new CatchUpDeamon(eventsManager,
				cState, conf, aBook));
		catchUpDeamon.setName("catchUpDeamon");
		catchUpDeamon.start();
		reConnectThread.startThread();
		// initiate user server
		userHandlerInterface.setAddRequestDealer(new AddRequestDealer(cState,
				conf, aBook, eventsManager));
		tryBindUserServer(userHandlerInterface);
	}

	public EventsManager getEventsManager() {
		return eventsManager;
	}

	private ClientNetHandler connectNodes() throws NumberFormatException,
			Exception {
		NioSocketConfigure nsc = new NioSocketConfigure();
		nsc.setProcessorNum(1);
		nsc.setTcpNoDelay(true);
		nsc.setClearTimeoutSessionInterval(20000);
		nsc.setCycleRecieveBuffCellSize(2000);
		NioSocketClient client = new NioSocketClient(nsc, "client");
		ClientNetHandler phandler = new ClientNetHandler(client, eventsManager,
				reConnectThread);
		IoHandler handler = new MultiThreadProtecolHandler(1000, 16 * 1024, 5,
				300, 60, TimeUnit.SECONDS, ProtobufProtocolBuilder.build(),
				phandler, "client");
		phandler.setHandler(handler);

		NodesCollection senators = cState.getSenators();
		NodesCollection followers = cState.getFollowers();
		for (Entry<String, NodeState> s : senators.getNodeStates().entrySet()) {
			String[] temp = s.getKey().split(":");
			String ip = temp[0];
			String port = temp[1];
			client.connect(new InetSocketAddress(ip, Integer.parseInt(port)),
					handler, s.getValue());
		}
		for (Entry<String, NodeState> s : followers.getNodeStates().entrySet()) {
			String[] temp = s.getKey().split(":");
			String ip = temp[0];
			String port = temp[1];
			client.connect(new InetSocketAddress(ip, Integer.parseInt(port)),
					handler, s.getValue());
		}
		return phandler;
	}

	private ServerNetHandler tryBind() throws SocketException {
		NioSocketConfigure nsc = new NioSocketConfigure();
		nsc.setTcpNoDelay(true);
		nsc.setProcessorNum(1);
		nsc.setClearTimeoutSessionInterval(20000);
		nsc.setCycleRecieveBuffCellSize(2000);

		HashSet<String> allIps = new HashSet<String>();
		Enumeration<NetworkInterface> interfaceEnum = NetworkInterface
				.getNetworkInterfaces();
		while (interfaceEnum.hasMoreElements()) {
			NetworkInterface i = interfaceEnum.nextElement();
			Enumeration<InetAddress> ipEnum = i.getInetAddresses();
			while (ipEnum.hasMoreElements()) {
				allIps.add(ipEnum.nextElement().getHostAddress());
			}
		}
		Set<ConfigNode> set = conf.getNodes();
		ServerNetHandler handler = new ServerNetHandler(eventsManager, conf);
		for (ConfigNode s : set) {
			if (allIps.contains(s.getIp())
					&& String.valueOf(conf.getPort()).equals(s.getPort())) {
				try {

					server = new NioSocketServer(nsc, "server");
					server.waitToBind(
							new InetSocketAddress(Integer.parseInt(s.getPort())),
							new MultiThreadProtecolHandler(1000, 16 * 1024, 5,
									300, 60, TimeUnit.SECONDS,
									ProtobufProtocolBuilder.build(), handler,
									"server"));
					conf.setSelfAddress(s.getIp() + ":" + s.getPort());
					return handler;
				} catch (IOException e) {
				}
			}
		}
		throw new RuntimeException(
				"fail to bind port,check configuration's file please.");
	}

	private UserServerHandler tryBindUserServer(
			UserHandlerInterface userHandlerInterface) throws SocketException {
		NioSocketConfigure nsc = new NioSocketConfigure();
		nsc.setTcpNoDelay(true);
		nsc.setProcessorNum(1);
		nsc.setClearTimeoutSessionInterval(20000);
		nsc.setCycleRecieveBuffCellSize(2000);

		HashSet<String> allIps = new HashSet<String>();
		Enumeration<NetworkInterface> interfaceEnum = NetworkInterface
				.getNetworkInterfaces();
		while (interfaceEnum.hasMoreElements()) {
			NetworkInterface i = interfaceEnum.nextElement();
			Enumeration<InetAddress> ipEnum = i.getInetAddresses();
			while (ipEnum.hasMoreElements()) {
				allIps.add(ipEnum.nextElement().getHostAddress());
			}
		}
		Set<ConfigNode> set = conf.getNodes();
		UserServerHandler handler = new UserServerHandler(eventsManager, conf,
				userHandlerInterface);
		for (ConfigNode s : set) {
			if (allIps.contains(s.getIp())
					&& String.valueOf(conf.getPort()).equals(s.getPort())) {
				try {

					server = new NioSocketServer(nsc, "userserver");
					server.waitToBind(
							new InetSocketAddress(Integer.parseInt(s
									.getClientPort())),
							new MultiThreadProtecolHandler(1000, 16 * 1024, 5,
									300, 60, TimeUnit.SECONDS,
									ProtobufProtocolBuilder.build(), handler,
									"userserver"));
					conf.setSelfAddress(s.getIp() + ":" + s.getPort());
					return handler;
				} catch (IOException e) {
				}
			}
		}
		throw new RuntimeException(
				"fail to bind user server port,check configuration's file please.");
	}
}
